/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.order;

import java.util.Hashtable;
import java.util.Iterator;
import java.util.TreeSet;

import cgl.narada.event.EventID;
import cgl.narada.service.qos.QosEvent;
import cgl.narada.service.qos.impl.QosServiceImpl;
import cgl.narada.service.reliable.events.RdArchivalServiceNotification;

/** This class ensures the exactly once delivery of events  
    
    @author Shrideep Pallickara
    $Date$
    $Revision$
*/



public class OrderedDeliveryServiceImpl {
  
  private Hashtable templatesAndSyncs;
  private Hashtable pendingEvents;
  private Hashtable pendingNotifications;
  
  private QosServiceImpl qosServiceImpl;
  private boolean debug = false;

  
  private String moduleName = "OrderedDeliveryServiceImpl: ";

  
  public OrderedDeliveryServiceImpl(QosServiceImpl qosServiceImpl) {
    this.qosServiceImpl = qosServiceImpl;

    templatesAndSyncs = new Hashtable();
    pendingEvents = new Hashtable();
    pendingNotifications = new Hashtable();
  }

  /** Number of events queued within the service pending to be released. The
      events are queued to satisfy certain ordering constraints */
  public int getNumberOfEventsPendingRelease() {
    return pendingEvents.size();
  }

  /** Procees the advancement of sync point for a given templateId and entityId
   */
  public void 
  onAdvanceOfSyncpoint(int entityId, int templateId, long advanceSync) {
    Object templateKey = new Integer(templateId);
    
    if (templatesAndSyncs.containsKey(templateKey) ) {
      long oldSync = ((Long)templatesAndSyncs.get(templateKey)).longValue();
      if (oldSync < advanceSync) {
	templatesAndSyncs.remove(templateKey);
      } else {
	/** No need to process this advancement, if it is less or equal to
	    the current advancement */
	return;
      }
    }

    templatesAndSyncs.put(templateKey, new Long(advanceSync));
    /** Check to see if any of the pending events can be released as a
	result of this syncpoint advancement */
    
    tryAndReleaseOrderedEvents(entityId, templateId, advanceSync);
  }
  
  
  


  public void 
  onReliableDelivery(QosEvent qosEvent, 
		     RdArchivalServiceNotification archNotification) {
    // throws ServiceException {
    EventID eventId = qosEvent.getEventId();
    
    if (!archNotification.getEventId().equals(eventId)) {
      //throw 
      //new ServiceException(moduleName + "The event ID assoicated with "+
      System.out.println(moduleName + "The event ID assoicated with " +
			 "the QosEvent =" + eventId + " is different " +
			 "from the one associated with the archival " +
			 "notification =" + archNotification.getEventId());
    }
    
    if (pendingEvents.contains(eventId)) {
      /** Event was previously queued ... */
      return;
    }
    
    int templateId = archNotification.getTemplateId();

    /** Check to see if the event was released previously */
    Object templateKey = new Integer(templateId);
    if (templatesAndSyncs.containsKey(templateKey) ) {
      long currentSync = 
	((Long)templatesAndSyncs.get(templateKey)).longValue();
      long archSync = archNotification.getSequenceNumber();
      
      if (currentSync >= archSync) {
	return;
      }       
    }

    /** Proceed to add the event and the archival notification into the
	corresponding pending queues */
    pendingEvents.put(eventId, qosEvent);

    if (!pendingNotifications.containsKey(templateKey)) {
      pendingNotifications.put(templateKey, new TreeSet(archNotification));
    }
    
    TreeSet archSet = (TreeSet) pendingNotifications.get(templateKey);
    archSet.add(archNotification);
    System.out.println(moduleName + "Total number of pending events =" +
		       pendingEvents.size() + ", sequence number of latest " +
		       "addition =" + archNotification.getSequenceNumber() );
  }

  private void 
  tryAndReleaseOrderedEvents(int entityId, int templateId, long advanceSync) {
    Object templateKey = new Integer(templateId);
    TreeSet archSet = (TreeSet) pendingNotifications.get(templateKey);
    
    if (archSet == null) {
      if (debug) {
	System.out.println(moduleName + "No pending events/notifications");
      }
      return;
    }
    System.out.println(moduleName + "EntityId=" + entityId + ", TemplateId=" +
		       templateId + ", Advance Sync=" + advanceSync);

    Iterator iter = archSet.iterator();
    int numProcessed = 0;
    while (iter.hasNext()) {
      RdArchivalServiceNotification archNotification =
	(RdArchivalServiceNotification) iter.next();
      
      if (archNotification.getSequenceNumber() <= advanceSync ) {
	EventID eventId = archNotification.getEventId();
	QosEvent qosEvent = (QosEvent) pendingEvents.remove(eventId);
	System.out.println(moduleName + 
			   "Proceeding to release ordered event with sync=" + 
			   archNotification.getSequenceNumber());
	numProcessed++;
	//archSet.remove(archNotification);
	qosServiceImpl.onOrderedDelivery(qosEvent);
      } else {
	/** The set is a sorted one, so the first time we encounter a 
	    value which is greater than the sync advance we exit from
	    the loop, since all the remaining values will continue to be
	    higher */
	break;
      } /* end if */      
    }/* end while*/
    
    for (int i=0; i< numProcessed ; i++) {
      Object obj = archSet.first();
      archSet.remove(obj);
    }

  }/* end tryAndReleaseOrderedEvents() */


}
